package com.sunzm.spark.sql.hive.report.dwd

import org.apache.spark.sql.SparkSession

/**
 *
 * 用户拉链表存在的意义：
 *
 * 假设 2021-07-01的数据如下：
 * 1,张三, 2021-07-01
 * 2021-07-02的数据如下：
 * 1,张三, 2021-07-02
 * 2021-07-03的数据如下：
 * 1,张三2, 2021-07-03
 *
 * 可以发现， 1,张三 这条数据存了2次，如果不考虑空间占用的问题，这样保存数据是没问题的。
 *
 * 但是，假设有1000万条数据，每天只有十几个用户的信息发生改变。显然，这样存是浪费的。
 *
 * 我们可以使用一个拉链表，这样存数据：
 * 2021-07-01的数据如下：
 * * 1,张三, 2021-07-01,9999-12-31
 * * 2021-07-02的数据如下：
 * * 1,张三, 2021-07-01,9999-12-31
 * * 2021-07-03的数据如下(姓名发生变化了)：
 * * 1,张三,  2021-07-01,2021-07-02
 * * 1,张三2, 2021-07-03,9999-12-31
 * 看起来好像没有变化?
 * 其实区别在于，之前是7月1号，2号，3号都有一张表，存着这个用户的信息
 * 而拉链表是只有一张表，存着用户的信息有效时间段，也就是说拉链表里，有这个用户各个时间段的信息
 *
 * 那么，如果想要查询 2021-07-01号当时的用户信息，怎么查呢？
 * WHERE 区间开始时间<=2021-07-01  AND 区间结束时间 >= 2021-07-01
 * @author Administrator
 * @version 1.0
 * @date 2021-07-30 0:08
 */
object DWDUserZipTable {
  def main(args: Array[String]): Unit = {
    val warehouseLocation = "/user/hive/warehouse"
    //val warehouseLocation = new File("spark-warehouse").getAbsolutePath

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.stripSuffix("$"))
      .master("local[*]")
      .config("spark.default.parallelism", 8)
      .config("spark.sql.shuffle.partitions", 8)
      //如果集群中有hive相关的配置文件，下面的都可以不写
      //.config("fs.defaultFS", "hdfs://192.168.1.158:8020")
      .config("fs.defaultFS", "file:///")
      .config("spark.sql.warehouse.dir", warehouseLocation)
      //如果不开启对Hive的支持,元数据保存在内存中,程序退出,元数据也就丢了
      .enableHiveSupport()
      //开启动态分区支持
      .config("hive.exec.dynamic.partition", "true")
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("javax.jdo.option.ConnectionDriverName", "com.mysql.jdbc.Driver")
      .config("javax.jdo.option.ConnectionURL", "jdbc:mysql://82.156.210.70:3306/hive?useSSL=false")
      .config("javax.jdo.option.ConnectionUserName", "root")
      .config("javax.jdo.option.ConnectionPassword", "ABC123abc.123")
      //高版本的Hive需要设置下面的参数
      .config("hive.metastore.schema.verification", false)
      .config("datanucleus.schema.autoCreateTables", true)
      .getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._

    //准备测试数据
    //拉链表
    val seq = Seq(
      //用户ID, 用户名, 起始时间, 结束时间
      (1, "张三", "2021-01-01", "9999-12-31"),
      (2, "李四", "2021-05-01", "2021-06-01"),
      (2, "李四2", "2021-06-02", "9999-12-31")
    )

    spark.createDataset(seq)
      .toDF("user_id", "user_name", "start_date", "end_date")
      .createOrReplaceTempView("v_app_user_zip")

    //今日用户增量表（这个数据一般是mysql导入hive的，今天修改或者新增的就是增量数据，起始就是变化和新增数据）
    val userSeq = Seq(
      //用户ID, 用户名, 时间
      (2, "李四3", "2021-07-30"),
      (3, "王五", "2021-07-30")
    )

    spark.createDataset(userSeq)
      .toDF("user_id", "user_name", "op_date")
      .createOrReplaceTempView("v_app_user_ac")

    //两张表JOIN，形成新的拉链表
    //如果右表没有，说明没有变化过，直接取左表的数据
    //如果右表也有，说明是新增的数据，需要把结束时间修改为昨天
    //最后，需要再UNION上新增的数据，新增的左连接是没有的
    val zipDF = spark.sql(
      """
        | SELECT
        |   t1.user_id,
        |   t1.user_name,
        |   t1.start_date,
        |   IF(t2.user_id IS NULL OR t1.end_date <> '9999-12-31', t1.end_date, '2021-07-29') AS end_date
        | FROM v_app_user_zip t1 LEFT JOIN v_app_user_ac t2
        |   ON t1.user_id = t2.user_id
        | UNION ALL
        | SELECT
        |   user_id,
        |   user_name,
        |   '2021-07-30' AS start_date,
        |   '9999-12-31' AS end_date
        | FROM v_app_user_ac
        |""".stripMargin)

    zipDF.show(10, false)
    zipDF.createOrReplaceTempView("v_app_user_zip_new")

    //查询最新的用户信息
    spark.sql(
      """
        |SELECT
        | user_id,
        | user_name
        |FROM v_app_user_zip_new
        | WHERE end_date = '9999-12-31'
        |""".stripMargin)
      .show(10, false)

    //查询某一天的用户信息
    spark.sql(
      """
        |SELECT
        | user_id,
        | user_name
        |FROM v_app_user_zip_new
        | WHERE start_date <= '2021-06-30'
        | AND end_date >= '2021-06-30'
        |""".stripMargin)
      .show(10, false)

    spark.stop()
  }
}
